package pool;


import java.util.ArrayDeque;
import java.util.Deque;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 阻塞队列
 *
 * @author yezhiqiu
 * @date 2021/10/01
 */
public class BlockingQueue<T> {

    /**
     * 队列
     */
    private final Deque<T> deque;

    /**
     * 容量
     */
    private final Integer capacity;

    /**
     * 锁
     */
    private final ReentrantLock lock = new ReentrantLock();

    /**
     * 生产者条件
     */
    Condition produceCondition = lock.newCondition();

    /**
     * 消费者条件
     */
    Condition consumerCondition = lock.newCondition();

    public BlockingQueue(Integer capacity) {
        this.capacity = capacity;
        this.deque = new ArrayDeque<>(capacity);
    }

    /**
     * 添加任务
     *
     * @param task 元素
     */
    public void put(T task) {
        lock.lock();
        try {
            // 判断队列是否已满
            while (deque.size() == capacity) {
                try {
                    produceCondition.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            deque.addLast(task);
            // 通知消费者消费
            consumerCondition.signal();
        } finally {
            lock.unlock();
        }
    }

    /**
     * 判断队列是否已满
     *
     * @return {@link Boolean}
     */
    public Boolean isFull() {
        lock.lock();
        try {
            return deque.size() == capacity;
        } finally {
            lock.unlock();
        }
    }

    /**
     * 添加任务
     *
     * @param rejectedExecutionHandler 拒绝策略
     * @param task         任务
     */
    public void tryPut(RejectedExecutionHandler<T> rejectedExecutionHandler, T task) {
        lock.lock();
        try {
            // 判断队列是否已满
            if (isFull()) {
                rejectedExecutionHandler.reject(this, task);
            } else {
                deque.addLast(task);
                // 通知消费者消费
                consumerCondition.signal();
            }
        } finally {
            lock.unlock();
        }
    }

    /**
     * 限定时间内添加任务
     *
     * @param task     任务
     * @param timeout  超时
     * @param timeUnit 时间单位
     * @return {@link Boolean}
     */
    public Boolean offer(T task, Long timeout, TimeUnit timeUnit) {
        lock.lock();
        // 转换纳秒单位
        timeout = timeUnit.toNanos(timeout);
        try {
            // 判断队列是否已满
            while (deque.size() == capacity) {
                try {
                    if (timeout <= 0) {
                        return false;
                    }
                    timeout = produceCondition.awaitNanos(timeout);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            deque.addLast(task);
            // 通知消费者消费
            consumerCondition.signal();
            return true;
        } finally {
            lock.unlock();
        }
    }

    /**
     * 取出任务
     *
     * @return {@link T} 任务
     */
    public T take() {
        lock.lock();
        try {
            // 判断队列是否为空
            while (deque.isEmpty()) {
                try {
                    consumerCondition.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            // 通知消费者消费
            consumerCondition.signal();
            return deque.removeFirst();
        } finally {
            lock.unlock();
        }
    }

    /**
     * 限定时间内取出任务
     *
     * @param timeout  超时
     * @param timeUnit 时间单位
     * @return {@link T}
     */
    public T poll(Long timeout, TimeUnit timeUnit) {
        lock.lock();
        try {
            // 转换纳秒单位
            timeout = timeUnit.toNanos(timeout);
            // 判断队列是否为空
            while (deque.isEmpty()) {
                try {
                    // 超时直接返回
                    if (timeout <= 0) {
                        return null;
                    }
                    // 获取等待时间
                    timeout = consumerCondition.awaitNanos(timeout);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            // 通知生产者生产
            produceCondition.signal();
            return deque.removeFirst();
        } finally {
            lock.unlock();
        }
    }

}
